package com.yanggu.market_analysis

import cn.hutool.core.date.DateUtil
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.concurrent.TimeUnit
import java.util.{Date, UUID}
import scala.util.Random

/**
 * 统计过去1分钟内渠道和行为的分组数量, 5秒钟输出一次
 */
object AppMarketingByChannel {

  def main(args: Array[String]): Unit = {

    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    environment
      .addSource(new SimulatedEventSource)
      .filter(_.behavior != "UNINSTALL")
      .assignAscendingTimestamps(_.timestamp)
      .map(data => (data.channel, data.behavior))
      .keyBy(data => data)
      .window(SlidingEventTimeWindows.of(Time.minutes(1L), Time.seconds(5L)))
      //process方法在Watermark大于等于windowEnd的时间戳的时候就会执行
      .process(new MarketingCountByChannel)
      .print("result")

    environment.execute("AppMarketingByChannel Job")

  }

}

/**
 * 市场分析用户行为样例类
 *
 * @param userId    用户id
 * @param behavior  具体的行为
 * @param channel   渠道
 * @param timestamp 时间戳
 */
case class MarketingUserBehavior(userId: String, behavior: String, channel: String, timestamp: Long)

case class MarketingCountView(windowStart: String, windowEnd: String, channel: String, behavior: String, count: Long)

class SimulatedEventSource extends RichParallelSourceFunction[MarketingUserBehavior] {

  var running = true

  private val channelList = List[String]("AppStore", "XiaomiStore", "HuaweiStore", "weibo", "wechat", "tieba")

  private val behaviorTypes = List[String]("BROWSE", "CLICK", "PURCHASE", "UNINSTALL")

  private val random = Random

  override def run(ctx: SourceFunction.SourceContext[MarketingUserBehavior]): Unit = {
    val behaviorTypeSize = behaviorTypes.size
    val channelListSize = channelList.size

    while (running) {
      val userId = UUID.randomUUID().toString
      val behavior = behaviorTypes(random.nextInt(behaviorTypeSize))
      val channel = channelList(random.nextInt(channelListSize))
      val ts = System.currentTimeMillis()
      ctx.collect(MarketingUserBehavior(userId, behavior, channel, ts))
      //睡眠5毫秒
      TimeUnit.MILLISECONDS.sleep(5L)
    }

  }

  override def cancel(): Unit = running = false

}

class MarketingCountByChannel extends ProcessWindowFunction[(String, String), MarketingCountView, (String, String), TimeWindow] {

  //elements是窗口内的所有数据
  override def process(key: (String, String),
                       context: Context,
                       elements: Iterable[(String, String)],
                       out: Collector[MarketingCountView]): Unit = {
    out.collect(MarketingCountView(DateUtil.formatDateTime(new Date(context.window.getStart)),
      DateUtil.formatDateTime(new Date(context.window.getEnd)),
      key._1, key._2, elements.size))
  }

}